package com.tim.helper;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.junit.Before;
import org.junit.Test;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Date: 2023/9/19 10:13
 *
 * @Author: Tim
 * @Description:
 */
public class RocketmqHelper {
    String nameServer = "";
    String accessKey = "";
    String secretKey = "";
    String topic = "";
    String group = "";
    String tag = "";
    int times = 1;
    int sendBatchSize=10;
    Object content = "{\n" +
            "  \"name\": \"demoData\"\n" +
            "}";
    ExecutorService executorService = Executors.newFixedThreadPool(10);

    @Before
    public void init(){
        /**/
        {
            // dev
            nameServer = "";
            accessKey = "";
            secretKey = "";

            // sit
            nameServer = "";
            accessKey = "";
            secretKey = "";

            // uat
            nameServer = "";
            accessKey = "";
            secretKey = "";

        }
    }

    @Test
    public void sendOne() throws Exception {
        DefaultMQProducer mqProducer = getMQProducer();

        Message msg = new Message(topic, tag, JSON.toJSONString(content).getBytes());
        SendResult send = mqProducer.send(msg);
        System.out.println(JSON.toJSONString(send));
        mqProducer.shutdown();
    }

    @Test
    public void sendMulti() {
        DefaultMQProducer mqProducer = getMQProducer();
        List<Message> list = new ArrayList<>(sendBatchSize);
        for (int i = 1; i <= times; i++) {
            content = generateContent();
            if(content==null){
                throw new RuntimeException("content不能为null");
            }
            Message msg = new Message(topic, tag, JSON.toJSONString(content).getBytes());
            list.add(msg);
            if(list.size()%sendBatchSize==0){
                executorService.submit(()->{
                    SendResult send = null;
                    try {
                        send = mqProducer.send(list);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                    System.out.println(JSON.toJSONString(send));
                });
                list.clear();
            }
        }
        if(!list.isEmpty()){
            executorService.submit(()->{
                SendResult send = null;
                try {
                    send = mqProducer.send(list);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
                System.out.println(JSON.toJSONString(send));
            });
        }
        executorService.shutdown();
        while (!executorService.isShutdown()){

        }
        mqProducer.shutdown();
    }

    /**producer可向所有topic发送消息*/
    public DefaultMQProducer getMQProducer() {
        RocketMQProperties rocketMQProperties = new RocketMQProperties();
        rocketMQProperties.setNameServer(nameServer);
        RocketMQProperties.Producer producerConfig = new RocketMQProperties.Producer();
        producerConfig.setGroup(group);
        producerConfig.setAccessKey(accessKey);
        producerConfig.setSecretKey(secretKey);
        rocketMQProperties.setProducer(producerConfig);
        String nameServer = rocketMQProperties.getNameServer();
        String groupName = producerConfig.getGroup();
        Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
        Assert.hasText(groupName, "[rocketmq.producer.group] must not be null");
        String accessChannel = rocketMQProperties.getAccessChannel();
        String ak = producerConfig.getAccessKey();
        String sk = producerConfig.getSecretKey();
        boolean isEnableMsgTrace = producerConfig.isEnableMsgTrace();
        String customizedTraceTopic = producerConfig.getCustomizedTraceTopic();
        DefaultMQProducer producer = RocketMQUtil.createDefaultMQProducer(groupName, ak, sk, isEnableMsgTrace, customizedTraceTopic);
        producer.setNamesrvAddr(nameServer);
        if (StringUtils.hasLength(accessChannel)) {
            producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
        }

        producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
        producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
        producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
        producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
        producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
        producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
        producer.setUseTLS(producerConfig.isTlsEnable());
        producer.setNamespace(producerConfig.getNamespace());
        producer.setInstanceName(producerConfig.getInstanceName());

        try {
            producer.start();
        } catch (MQClientException var12) {
            var12.printStackTrace();
            throw new RuntimeException("defaultLitePullConsumer is run fail");
        }

        return producer;
    }

    public Object generateContent(){

        return null;
    }
}
